package com.Hsu.mq.mqserver;

/**
 * @author Hsu琛君珩
 * @ClassName:VirtualHost
 * @date 2024-02-20
 * @apiNote
 * @Version: v1.0
 */

import com.Hsu.mq.common.Consumer;
import com.Hsu.mq.common.MqException;
import com.Hsu.mq.mqserver.core.*;
import com.Hsu.mq.mqserver.datacenter.DiskDataCenter;
import com.Hsu.mq.mqserver.datacenter.MemoryDataCenter;
import lombok.Data;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 通过这个类表示虚拟主机
 * 每个虚拟主机下面都管理者自己的 交换机、队列、绑定、消息 数据，同时提供 api 供上层调用
 * 针对 VirtualHost 这个类，作为业务逻辑的整合者，就需要对于代码中抛出的异常进行处理了
 */
@Data
public class VirtualHost {
    private String virtualHostName;
    private MemoryDataCenter memoryDataCenter=new MemoryDataCenter();
    private DiskDataCenter diskDataCenter=new DiskDataCenter();
    private Router router=new Router();
    private ConsumerManager consumerManager=new ConsumerManager(this);

    //锁对象，操作交换机的锁对象
    private final Object exchangeLocker=new Object();
    //操作队列的锁对象
    private final Object queueLocker=new Object();

    public VirtualHost(String name){
        this.virtualHostName=name;
        //对于 MemoryDataCenter 不需要额外的初始化，只要 new 出来就都有了
        //但对于 DiskDataCenter 还需要初始化，建库建表还有初始数据的设定
        diskDataCenter.init();
        //另外还需要针对硬盘的数据，进行恢复到内存中
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException | MqException | ClassNotFoundException e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 恢复内存数据失败！");
        }
    }

    /**
     * 创建交换机
     * 如果交换机不存在就创建，存在就直接返回
     * @param exchangeName
     * @param exchangeType
     * @param durable
     * @param autoDelete
     * @param arguments
     * @return 创建成功返回 true，失败返回 false
     */
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete, Map<String, Object> arguments){
        //把交换机的名字，加上虚拟主机作为前缀
        exchangeName=virtualHostName+exchangeName;
        try {
            synchronized (exchangeLocker){
                //1.判断该交换机是否已经存在，直接通过内存查询，硬盘中存储只是为了持久化
                Exchange existsExchange=memoryDataCenter.getExchange(exchangeName);
                if(existsExchange!=null){
                    //已经存在
                    System.out.println("[VirtualHost] 交换机已经存在！exchangeName="+exchangeName);
                    return true;
                }
                //2.真正创建交换机，先构造 Exchange 对象
                Exchange exchange=new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangeType);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);
                //3.把交换机对象写入硬盘
                if(durable){//是否需要持久化
                    diskDataCenter.insertExchange(exchange);
                }
                //4.把交换机对象写入内存
                memoryDataCenter.insertExchange(exchange);
                System.out.println("[VirtualHost] 交换机创建完成！exchangeName="+exchangeName);
                //上述逻辑先写硬盘再写内存，目的就是因为硬盘更容易写失败，如果硬盘写失败内存就不写了
                //如果先写内存，内存写成功了，硬盘写失败了，还需要把内存的数据删除掉，就比较麻烦
            }
            return true;
        }catch (Exception e){
            e.printStackTrace();
            System.out.println("[VirtualHost] 交换机创建失败！exchangeName="+exchangeName);
            return false;
        }
    }

    /**
     * 删除交换机
     * @param exchangeName
     * @return
     */
    public boolean exchangeDelete(String exchangeName){
        exchangeName=virtualHostName+exchangeName;
        try {
            synchronized (exchangeLocker){
                //1.先找到对应的交换机
                Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                if(exchange==null){
                    throw new MqException("[VirtualHost] 交换机不存在无法删除！");
                }
                //2.删除硬盘上的数据
                if(exchange.isDurable()){
                    diskDataCenter.deleteExchange(exchangeName);
                }
                //3.删除内存中的交换机数据
                memoryDataCenter.deleteExchange(exchangeName);
                System.out.println("[VirtualHost] 交换机删除成功！exchangeName="+exchangeName);
            }
            return true;
        }catch (Exception e){
            e.printStackTrace();
            System.out.println("[VirtualHost] 交换机删除失败！exchangeName="+exchangeName);
            return false;
        }
    }

    /**
     * 创建队列
     * @param queueName
     * @param durable
     * @param exclusive
     * @param autoDelete
     * @param arguments
     * @return
     */
    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments){
        //先把队列的名字拼接上虚拟主机的名字
        queueName=virtualHostName+queueName;
        try {
            synchronized (queueLocker){
                //1.判断队列是否存在
                MSGQueue existsQueue=memoryDataCenter.getQueue(queueName);
                if(existsQueue!=null){
                    System.out.println("[VirtualHost] 队列已经存在！queueName="+queueName);
                    return true;
                }
                //2.创建队列对象
                MSGQueue queue=new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);
                //3.把队列写入到硬盘
                if(durable){
                    diskDataCenter.insertQueue(queue);
                }
                //4.把队列写到内存
                memoryDataCenter.insertQueue(queue);
                System.out.println("[VirtualHost] 队列创建成功！queueName="+queueName);
            }
            return true;
        }catch (Exception e){
            e.printStackTrace();
            System.out.println("[VirtualHost] 队列创建失败！queueName="+queueName);
            return false;
        }
    }

    /**
     * 删除队列
     * @param queueName
     * @return
     */
    public boolean queueDelete(String queueName){
        queueName=virtualHostName+queueName;
        try {
            synchronized (queueLocker){
                //1.根据队列名字，查找当前队列对象
                MSGQueue queue=memoryDataCenter.getQueue(queueName);
                if(queue==null){
                    throw new MqException("[VirtualHost] 队列不存在！无法删除！queueName="+ queueName);
                }
                //2.删除硬盘数据
                if(queue.isDurable()){
                    diskDataCenter.deleteQueue(queueName);
                }
                //3.删除内存数据
                memoryDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualHost] 删除队列成功！queueName="+queueName);
            }
            return true;
        }catch (Exception e){
            e.printStackTrace();
            System.out.println("[VirtualHost] 删除队列失败！queueName="+queueName);
            return false;
        }
    }

    /**
     * 创建绑定
     * @param queueName
     * @param exchangeName
     * @param bindingKey
     * @return
     */
    public boolean queueBind(String queueName, String exchangeName, String bindingKey){
        queueName=virtualHostName+queueName;
        exchangeName=virtualHostName+exchangeName;
        try {
            synchronized (exchangeLocker){
                synchronized (queueLocker){
                    //1.判断当前的绑定是否已经存在
                    Binding existsBinding=memoryDataCenter.getBinding(exchangeName,queueName);
                    if(existsBinding!=null){
                        throw new MqException("[VirtualHost] 绑定已经存在！queueName="+queueName+", exchangeName="+exchangeName);
                    }
                    //2.验证 bindingKey 是否合法
                    if(!router.checkBindingKey(bindingKey)){
                        throw new MqException("[VirtualHost] bindingKey 非法！bindingKey="+bindingKey);
                    }
                    //3.创建绑定对象
                    Binding binding=new Binding();
                    binding.setExchangeName(exchangeName);
                    binding.setQueueName(queueName);
                    binding.setBindingKey(bindingKey);
                    //4.获取对应的交换机和队列，如果交换机或者队列不存在，这样的绑定也是无法创建的
                    MSGQueue queue=memoryDataCenter.getQueue(queueName);
                    if(queue==null){
                        throw new MqException("[VirtualHost] 队列不存在！queueName="+queueName);
                    }
                    Exchange exchange=memoryDataCenter.getExchange(exchangeName);
                    if(exchange==null){
                        throw new MqException("[VirtualHost] 交换机不存在！exchangeName="+exchangeName);
                    }
                    //5.先写硬盘
                    if(queue.isDurable()&&exchange.isDurable()){
                        diskDataCenter.insertBinding(binding);
                    }
                    //6.再写内存
                    memoryDataCenter.insertBinding(binding);
                    System.out.println("[VirtualHost] 绑定创建成功！exchangeName="+exchangeName+", queueName="+queueName);
                }
            }
            return true;
        }catch (Exception e){
            e.printStackTrace();
            System.out.println("[VirtualHost] 创建绑定失败！exchangeName="+exchangeName+", queueName="+queueName);
            return false;
        }
    }

    /**
     * 删除绑定
     * @param queueName
     * @param exchangeName
     * @return
     */
    public boolean queueUnbind(String queueName, String exchangeName){
        queueName=virtualHostName+queueName;
        exchangeName=virtualHostName+exchangeName;
        try {
            synchronized (exchangeLocker){
                synchronized (queueLocker){
                    //1.获取绑定，看看是否已经存在
                    Binding binding=memoryDataCenter.getBinding(exchangeName,queueName);
                    if(binding==null){
                        throw new MqException("[VirtualHost] 删除绑定失败！绑定不存在！exchangeName="+exchangeName+", queueName="+queueName);
                    }
                    ////2.获取对应的交换机和队列，看看是否存在
                    //MSGQueue queue=memoryDataCenter.getQueue(queueName);
                    //if(queue==null){
                    //    throw new MqException("[VirtualHost] 对应的队列不存在！queueName="+queueName);
                    //}
                    //Exchange exchange=memoryDataCenter.getExchange(exchangeName);
                    //if(exchange==null){
                    //    throw new MqException("[VirtualHost] 对应的交换机不存在！exchangeName="+exchangeName);
                    //}
                    //3.删除硬盘上的数据。无论绑定是否持久化，都从硬盘上删一下，就算不存在，也无副作用
                    //if(queue.isDurable()&&exchange.isDurable()){
                    diskDataCenter.deleteBinding(binding);
                    //}
                    //4.删除内存中的数据
                    memoryDataCenter.deleteBinding(binding);
                    System.out.println("[VirtualHost] 删除绑定成功！");
                }
            }
            return true;
        }catch (Exception e){
            e.printStackTrace();
            System.out.println("[VirtualHost] 删除绑定失败！");
            return false;
        }
    }

    /**
     * 发送消息到指定的交换机/队列中
     * @param exchangeName
     * @param routingKey
     * @param basicProperties
     * @param body
     * @return
     */
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body){
        try {
            //1.转换交换机的名字
            exchangeName=virtualHostName+exchangeName;
            //2.检查 routingKey 是否合法
            if(!router.checkRoutingKey(routingKey)){
                throw new MqException("[VirtualHost] routingKey 非法！routingKey="+routingKey);
            }
            //3.查找交换机对象
            Exchange exchange=memoryDataCenter.getExchange(exchangeName);
            if(exchange==null){
                throw new MqException("[VirtualHost] 交换机不存在！exchangeName="+exchangeName);
            }
            //4.判定交换机的类型
            if(exchange.getType()==ExchangeType.DIRECT){
                //按照直接交换机的方式转发消息
                //直接以 routingKey 作为队列的名字，直接把消息写入到指定的队列中
                //此时，可以无视绑定关系
                String queueName=virtualHostName+routingKey;
                //5.构造消息对象
                Message message=Message.createMessageWithId(routingKey,basicProperties,body);
                //6.查找该队列名对应的对象
                MSGQueue queue=memoryDataCenter.getQueue(queueName);
                if(queue==null){
                    throw new MqException("[VirtualHost] 队列不存在！queueName="+queueName);
                }
                //7.队列存在，直接给队列中写入消息
                sendMessage(queue, message);
            }else {
                //按照 FANOUT 和 TOPIC 的方式来转发
                //5.找到该交换机关联的所有绑定，并遍历这些绑定
                ConcurrentHashMap<String, Binding> bindingsMap = memoryDataCenter.getBindings(exchangeName);
                for(Map.Entry<String, Binding> entry:bindingsMap.entrySet()){
                    //1)获取到绑定对象，判定对应的队列是否存在
                    Binding binding=entry.getValue();
                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if(queue==null){
                        //此处可能有多个这样的队列，就不抛出异常了，不要因为一个队列的失败，影响到其他队列的消息的传输
                        System.out.println("[VirtualHost] basicPublish 发送消息时，发现队列不存在！queueName"+binding.getQueueName());
                        continue;
                    }
                    //2)构造消息对象
                    Message message=Message.createMessageWithId(routingKey,basicProperties,body);
                    //3)判定这个消息是否能转发给这个队列
                    //  如果是 FANOUT，所有绑定的队列都要转发的
                    //  如果是 TOPIC，还需要判定下，bindingKey 和 routingKey 是否匹配
                    if(!router.route(exchange.getType(),binding,message)){
                        continue;
                    }
                    //4)真正转发消息给队列
                    sendMessage(queue,message);
                }
            }
            return true;
        }catch (Exception e){
            e.printStackTrace();
            System.out.println("[VirtualHost] 消息发送失败！");
            return false;
        }
    }
    private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {
        //此处发送消息，就是把消息写入到硬盘和内存中
        int deliverMode = message.getDeliverMode();
        //deliverMode 为 1 就不持久化，为 2 就持久化
        if(deliverMode==2){
            diskDataCenter.sendMessage(queue,message);
        }
        //写入内存
        memoryDataCenter.sendMessage(queue,message);

        //此处还需要补充说明，通知消费者可以消费消息了
        consumerManager.notifyConsume(queue.getName());
    }

    /**
     * 订阅消息
     * 添加一个队列的订阅者，当队列收到消息之后，就要把消息推送给对应的订阅者
     * 消费者调用 basicConsume 就是订阅某个指定队列的消息
     * @param consumerTag 消费者的身份标识
     * @param queueName
     * @param autoAck 消费被消费完成后，应答的方式，true 为自动应答，false 为手动应答
     * @param consumer 函数式接口，这是一个回调函数，后续调用 basicConsume 并且传实参的时候，就可以写作 lambda 的形式了
     * @return
     */
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer){
        //构造一个 ConsumerEnv 对象，把这个对应的队列找到，再把这个 Consumer 对象添加到该队列中
        queueName=virtualHostName+queueName;
        try {
            consumerManager.addConsumer(consumerTag,queueName,autoAck,consumer);
            System.out.println("[VirtualHost] basicConsume 成功！queueName="+queueName);
            return true;
        }catch (Exception e){
            e.printStackTrace();
            System.out.println("[VirtualHost] basicConsume 失败！queueName="+queueName);
            return false;
        }
    }

    /**
     * 消息确认
     * @param queueName
     * @param messageId
     * @return
     */
    public boolean basicAck(String queueName, String messageId){
        queueName=virtualHostName+queueName;
        try {
            //1.获取消息和队列
            Message message=memoryDataCenter.getMessage(messageId);
            if(message==null){
                throw new MqException("[VirtualHost] 要确认的消息不存在！messageId="+messageId);
            }
            MSGQueue queue=memoryDataCenter.getQueue(queueName);
            if(queue==null){
                throw new MqException("[VirtualHost] 要确认的队列不存在！queueName="+queueName);
            }
            //2.删除硬盘上的数据
            if(message.getDeliverMode()==2){
                diskDataCenter.deleteMessage(queue,message);
            }
            //3.删除内存的消息中心的消息
            memoryDataCenter.removeMessage(messageId);
            //4.删除待确认集合的消息
            memoryDataCenter.removeMessageWaitAck(queueName,messageId);
            System.out.println("[VirtualHost] basicAck成功！消息被成功确认！queueName="+queueName+", messageId="+messageId);
            return true;
        }catch (Exception e){
            e.printStackTrace();
            System.out.println("[VirtualHost] basicAck失败！消息被成功失败！queueName="+queueName+", messageId="+messageId);
            return false;
        }
    }
}
